package com.zjl.SpringBoot.第21章_流式_响应式_编程.C_Reactive响应式核心;

import com.zjl.util.OutColour;
import org.junit.Test;
import reactor.core.publisher.Flux;

import java.io.IOException;
import java.util.*;

/**
 * ClassName: D2_Reactor实现之_Flux高级
 * Package: com.zjl.SpringBoot.第21章_流式_响应式_编程.C_Reactive响应式核心
 * Description:
 *
 * @Author 张蛟龙
 * @Create 2025/3/19 20:44
 * @Version 1.0
 */
public class D3_Flux高级方法 {
    @Test
    public void Flux响应式OnErrorXxx() throws IOException {

        //合并流
//        Flux.concat(Flux.just(1, 3, 5, 7), Flux.just(2, 4, 6))
        Flux<Integer> flux = Flux.range(1, 10)
                .map(i -> {
                    if (i == 3) {
                        i = i / 0;
                    }
                    return i;
                })//过滤
//                .onErrorComplete()//遇到错误后，之后都不识别，把错误吃掉
                .onErrorContinue((e, i) -> {//错误的数据走这里
                    OutColour.out.printlnRed("错误信息：", e.getMessage(), "\n错误的数据为：", i);
                });

        flux.subscribe(System.out::println);

        try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); }

    }

    @Test
    public void Flux响应式buffer() throws IOException {

        Flux.range(1, 10)
                .log()
                .limitRate(5)//预取限流   每次抓取指定个数据，如果 75% 元素已经处理了，继续抓取新的75%个元素
//                .buffer(3, () -> new ArrayList<>())//缓存3个      当1个使用    默认使用
                .buffer(3)//缓存3个      当1个使用    默认使用 ArrayList::new
//                .buffer(3,2)
//                .buffer()//默认所有当一批
                .subscribe(System.out::println)
                ;


        try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); }

    }
    @Test
    public void Flux响应式LimitRate() throws IOException {

        Flux.range(1, 10)
                .log()
                .limitRate(5)//预取限流   每次抓取指定个数据，如果 75% 元素已经处理了，继续抓取新的75%个元素
                .buffer(3)//缓存3个      当1个使用    默认使用 ArrayList::new
                .subscribe(System.out::println)
        ;


        try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); }

    }

}
